package ws;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import netty.util.StartAndStop;
import org.jctools.maps.NonBlockingHashSet;
import test.PerformanceAnalyze;
import ws.client.WebSocketClient;

import java.time.Instant;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Slf4j
public class OkHttpWsPublicOpenAPI extends StartAndStop {

    static PerformanceAnalyze performanceAnalyze = new PerformanceAnalyze();

    private void performanceAnalyze(JSONObject jsonObject) {
        if (jsonObject == null) {
            return;
        }
        long creationTime = jsonObject.getLongValue("creationTime");
        if (creationTime != 0) {
            performanceAnalyze.delay(System.currentTimeMillis() - creationTime);
        }
    }

    private void sleep(int sleep) {
        try {
            Thread.sleep(sleep);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void start(String url, int size, String channel, boolean enableCompress, int sleep) throws Exception {
        ScheduledExecutorService executor = Executors.newScheduledThreadPool(10);

        NonBlockingHashSet<WebSocketClient> wsList = new NonBlockingHashSet<>();

        CountDownLatch clientCounter = new CountDownLatch(size);

        for (int i = 0; i < size; i++) {
            executor.execute(() -> {
                try {
                    WebSocketClient webSocketClient = WebSocketClient.create(enableCompress, url, msg -> {
                        sleep(sleep);

                        JSONObject jsonObject = null;
                        try {
                            jsonObject = JSON.parseObject(msg);
//                            System.out.println(msg);
                        } catch (JSONException e) {
                            System.err.println(Instant.now());
                            System.err.println(msg);
                        }

                        performanceAnalyze(jsonObject);

                        performanceAnalyze.length(msg.length());
                        performanceAnalyze.meter();
                    });

                    wsList.add(webSocketClient);
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    clientCounter.countDown();
                }
            });
        }

        clientCounter.await(20, TimeUnit.SECONDS);

        for (WebSocketClient webSocket : wsList) {
            // {"op":"subscribe","id":"1645784808000","args":["1"]}
            // {"op":"fuzzySub","id":"1648539957000","args":["ticker"]}
            Map<String, Object> subscribe = new HashMap<>();
            subscribe.put("id", String.valueOf(System.currentTimeMillis()));
            subscribe.put("op", "subscribe");
            List<String> args = new LinkedList<>();
            args.add(channel);
            subscribe.put("args", args);
            String command = JSON.toJSONString(subscribe);
            webSocket.send(command);
        }

        for (WebSocketClient webSocket : wsList) {
            Map<String, Object> ping = new HashMap<>();
            ping.put("id", String.valueOf(System.currentTimeMillis()));
            ping.put("op", "ping");

            executor.scheduleAtFixedRate(() -> webSocket.send(JSON.toJSONString(ping)), 1, 10, TimeUnit.SECONDS);
        }
    }

}
